package com.lmm.rpc.core.handler;

import com.lmm.rpc.core.cache.LocalServerCache;
import com.lmm.rpc.core.client.protocol.MessageHeader;
import com.lmm.rpc.core.client.protocol.MessageProtocol;
import com.lmm.rpc.core.client.protocol.MessageStatus;
import com.lmm.rpc.core.client.protocol.MessageType;
import com.lmm.rpc.core.common.reponse.DefaultRepose;
import com.lmm.rpc.core.common.request.DefaultRequest;
import com.lmm.rpc.core.exception.RpcException;
import com.lmm.rpc.core.transport.MessageHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.concurrent.ExecutorService;

/**
 *
 * 服务端请求处理器
 * */
@Slf4j
public class DefaultRequestHandler extends SimpleChannelInboundHandler<MessageProtocol<DefaultRequest>> {
    //业务线程池
    private final ExecutorService threadPoolExecutor;
    //业务处理器
    private final MessageHandler messageHandler;
    public DefaultRequestHandler(ExecutorService threadPoolExecutor, MessageHandler messageHandler){
        this.threadPoolExecutor = threadPoolExecutor;
        this.messageHandler = messageHandler;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, MessageProtocol<DefaultRequest> messageProtocol) throws Exception {

        //处理心跳包 todo

        threadPoolExecutor.submit(()->{
            MessageProtocol<DefaultRepose> objectMessageProtocol = new MessageProtocol<>();
            DefaultRepose repose = new DefaultRepose();
            MessageHeader header = messageProtocol.getHeader();
            header.setMsgType(MessageType.RESPONSE.getType());
            try{
                if (messageHandler != null){
                    repose.setData(messageHandler.handle(messageProtocol.getBody()));
                }else{
                    repose.setData(handle(messageProtocol.getBody()));
                }

                header.setStatus(MessageStatus.SUCCESS.getCode());
                objectMessageProtocol.setHeader(header);
                objectMessageProtocol.setBody(repose);
            }catch (Throwable throwable){
                header.setStatus(MessageStatus.FAIL.getCode());
                repose.setMessage(throwable.toString());
                log.error("request deal error! requestId:{}",header.getRequestId(),throwable);
            }
            channelHandlerContext.writeAndFlush(objectMessageProtocol);
            log.info("send message to client!");
        });
    }

    /**
     * 通过反射来处理
     * */
    private Object handle(DefaultRequest request) {
        Object bean = LocalServerCache.get(request.getInterfaceName());
        if (bean == null){
            throw new RpcException(request.getInterfaceName()+" not exist!");
        }
        try {
            Method method = bean.getClass().getMethod(request.getMethodName(), request.getParameterTypes());
            return method.invoke(bean,request.getArguments());
        } catch (Exception e) {
            throw new RpcException(e);
        }
    }
}
